package com.llb666.shop.items.infracore.mq;

import com.llb666.common.core.utils.DateUtil;
import com.llb666.shop.items.domain.service.ItemsService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.function.Consumer;

@Component
@Slf4j
public class RabbitReceive {

    private final ItemsService itemsService;

    public RabbitReceive(ItemsService itemsService) {
        this.itemsService = itemsService;
    }

    // 可采用 ${spring.listener.xxx}的配置方式替换。
    @RabbitListener(bindings =
    @QueueBinding(
            value = @Queue(value = "order-queue"),
            exchange = @Exchange(name = "order-exchange", type = "topic", ignoreDeclarationExceptions = "true"),
            key = "order.*"))
    @RabbitHandler
    public void onMessage(Message<?> message, Channel channel) throws Exception {
        // 1,收到消息以后进行业务端消息消费处理
        log.info("消费消息：" + message.getPayload());
        // 2,处理成功后，获取deliveryTag,并进行手工ACK
        Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        log.info(String.valueOf(deliveryTag));
        channel.basicAck(deliveryTag, false);
    }


    /**
     * 消费分组消息
     *
     * @return message
     */
    @Bean
    public Consumer<Message<Map<String,Object>>> orderGroup() {
        log.info("ackMessage-初始化订阅");
        return obj -> {
            Channel channel = obj.getHeaders().get(AmqpHeaders.CHANNEL, Channel.class);
            Long deliveryTag = obj.getHeaders().get(AmqpHeaders.DELIVERY_TAG, Long.class);
            try{
                log.info("ackMessage-消息接收成功：" + obj.getPayload());
                Map<String, Object> payload = obj.getPayload();
                String itemSpecId = (String) payload.get("itemSpecId");
                int buyCounts = (int) payload.get("buyCounts");

                log.info("orderGroup接收分组消息：商品规格id为{},减少库存为{}", itemSpecId, buyCounts);
                itemsService.decreaseItemSpecStock(itemSpecId, buyCounts);
                // ack确认
                assert channel != null;
                channel.basicAck(deliveryTag, false);
            } catch (Exception e){
                log.error(e.getMessage());
            }
        };
    }

    //延时消息
//    @Bean
//    public Consumer<Message<String>> delay() {
//        return obj -> {
//            String msg = obj.getPayload();
//            log.info("delay-消息接收成功：" + DateUtil.dateToStringWithTime() + "  " + msg);
//        };
//    }

}
